package com.mazaiting.rabbitmq.config;

import com.mazaiting.common.core.utils.StringUtil;
import lombok.RequiredArgsConstructor;
import org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.retry.interceptor.RetryOperationsInterceptor;

/**
 * RabbitMQ 配置
 *
 * @author mazaiting
 * @create_time 2022/4/1 21:21
 */
@Configuration
@RequiredArgsConstructor
public class RabbitConfig {

    /**
     * 确认回调
     */
    private final RabbitTemplate.ConfirmCallback confirmCallback;

    /**
     * 消息返回回调
     */
    private final RabbitTemplate.ReturnsCallback returnsCallback;

    private final MqProperties mqProperties;

    /**
     * RabbitTemplate 配置
     * @return RabbitTemplate
     */
    @Bean
    @ConditionalOnMissingBean(RabbitTemplate.class)
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        // 设置确认回调
        rabbitTemplate.setConfirmCallback(confirmCallback);
        // 确保消息发送失败后可以重新返回到队列中, yml需要配置 publisher-returns: true
        rabbitTemplate.setMandatory(true);
        // 消息投递到队列失败回调处理
        rabbitTemplate.setReturnsCallback(returnsCallback);
        //使用单独的发送连接，避免生产者由于各种原因阻塞而导致消费者同样阻塞
        rabbitTemplate.setUsePublisherConnection(true);
        return rabbitTemplate;
    }

    /**
     * 消息发送容器
     *
     * @param connectionFactory 连接池工厂
     * @return 容器工厂
     */
    @Bean
    @ConditionalOnMissingBean(RabbitListenerContainerFactory.class)
    public RabbitListenerContainerFactory<DirectMessageListenerContainer> retryRabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        // 线程共享
        DirectRabbitListenerContainerFactory factory = new DirectRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        // 消息转换器
        factory.setMessageConverter(messageConverter());
        // 消息发送者发送的数量
        factory.setPrefetchCount(mqProperties.getPrefetchCount());
        // 为每个队列添加多个消费者 增加并行度
        factory.setConsumersPerQueue(mqProperties.getConsumersPerQueue());
        //设置消费者唯一标记 基于queue设置
        factory.setConsumerTagStrategy(queue -> queue + "_" + StringUtil.getUUID());
        // 设置手动确认模式
        factory.setAcknowledgeMode(mqProperties.getAckMode());
        // 设置重试机制
        factory.setAdviceChain(retryOperationsInterceptor());
        // 设置默认拒绝重回队列
        factory.setDefaultRequeueRejected(false);
        return factory;
    }

    /**
     * 管理者配置
     */
    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        // autoStartup 必须设置为 true, 否则 spring 容器不会加载 RabbitAdmin 类
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }


    /**
     * 重试机制 - 只有在自动确认的情况下生效
     *
     * @return 重试操作拦截器
     */
    @Bean
    public RetryOperationsInterceptor retryOperationsInterceptor() {
        return RetryInterceptorBuilder
                .stateless()
                //重试次数
                .maxAttempts(mqProperties.getMaxAttempts())
                //重试间隔  指数递增时间参数   最大间隔时间
                .backOffOptions(1000, 3, 10000)
                //次数用完之后的处理,用的是默认处理类,失败消息会到死信
                .recoverer(new RejectAndDontRequeueRecoverer())
//                .recoverer(new RepublishMessageRecoverer(rabbitTemplate))
                .build();
    }

    /**
     * 接受数据自动的转换为Json
     */
    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

}
